// -*-mode:c++; coding:utf-8-*-

#include "logger.hpp"
#include "naming_strategy.hpp"

#include "parallel_scanner.hpp"

namespace xbase
{
	typedef parallel_scanner::scan_state_map scan_state_map;

	parallel_scanner::parallel_scanner(io_service_pool &pool, session_manager &sm, location_cache &lc,
		const scan_info &si, handler_type handler)
		: base_type(pool),
		m_location_cache(lc),
		m_session_manager(sm),
		m_started(false),
		m_handler(handler),
		m_io_service_pool(pool),
		m_table_name(si.m_table_name)
	{
		m_tablet_scan_info.m_column_names = si.m_column_names;
		m_tablet_scan_info.m_latest_time_stamp = si.m_latest_time_stamp;
		m_tablet_scan_info.m_num_row = si.m_num_row;
		m_tablet_scan_info.m_oldest_time_stamp = si.m_oldest_time_stamp;
		m_tablet_scan_info.m_options = si.m_options;
		m_tablet_scan_info.m_row_count = si.m_row_count;
		m_tablet_scan_info.m_version_number = si.m_version_number;

		m_start_row = si.m_start_row;
		m_end_row = si.m_end_row;

		scan_state_map_value value;
		value.m_end_row = si.m_end_row;
		value.m_state = not_assigned;
		value.m_start_row_got = false;

		m_scan_state_map[si.m_start_row] = value;
	}

	void parallel_scanner::start()
	{
		if (!m_started)
		{
			m_started = true;

			// 先在缓存中查找
			m_location_cache.async_get_meta(
				this->get_lower_level_service().wrap(boost::bind(&this_type::on_get_meta, this_ptr(), _1, _2)),
				m_table_name, m_start_row, false, 0
				);
		}
	}

	void parallel_scanner::on_get_meta(location_cache_value meta, const error_code& ec)
	{
		LOG(TRACE, "on_get_meta ec=" << ec << " - " << ec.message());

		if (!ec)
		{
			LOG(DEBUG, "tablet: " << meta.tablet_info_p->m_table_name
				<< " [" << meta.tablet_info_p->m_start_row << "," << meta.tablet_info_p->m_end_row << "]"
				<< " is at " << meta.server_addr_p->m_hostname << ":" << meta.server_addr_p->m_port
				);

			endpoint_type ep = meta.server_addr_p->to_endpoint();

			std::string start_row = max(meta.tablet_info_p->m_start_row, m_start_row);
			std::string end_row;

			if (meta.tablet_info_p->m_end_row=="" || m_end_row=="")
			{
				end_row = (meta.tablet_info_p->m_end_row=="")?m_end_row:meta.tablet_info_p->m_end_row;
			}
			else
			{
				end_row = min(meta.tablet_info_p->m_end_row, m_end_row);
			}

			scan_state_map::iterator iter = m_scan_state_map.upper_bound(start_row);
			--iter;

			while (iter != m_scan_state_map.end()
				&& (end_row == "" || end_row > iter->first))
			{
				if (iter->second.m_state != not_assigned)
				{
					++iter;
				}
				else
				{
					if (iter->first < start_row)
					{
						// 拆分
						scan_state_map_value value;
						value.m_end_row = iter->second.m_end_row;
						value.m_state = not_assigned;
						value.m_start_row_got = false;

						iter->second.m_end_row = start_row;

						m_scan_state_map.insert(iter, scan_state_map::value_type(start_row, value));

						++iter;
					}

					if ((iter->second.m_end_row=="" && end_row!="")
						|| iter->second.m_end_row > end_row)
					{
						// 拆分
						scan_state_map_value value;
						value.m_end_row = iter->second.m_end_row;
						value.m_state = not_assigned;
						value.m_start_row_got = false;

						iter->second.m_end_row = end_row;

						m_scan_state_map.insert(iter, scan_state_map::value_type(end_row, value));
					}

					// 找寻可以连续扫描的一段
					std::string scan_start_row = iter->first;
					std::string scan_end_row = iter->second.m_end_row;

					scan_state_map::iterator scan_end_iter = iter;
					++scan_end_iter;

					while (scan_end_iter != m_scan_state_map.end()
						&& scan_end_iter->second.m_state == not_assigned
						&& (end_row == "" || scan_end_iter->first < end_row))
					{
						if ((scan_end_iter->second.m_end_row=="" && end_row!="")
							|| scan_end_iter->second.m_end_row > end_row)
						{
							// 拆分
							scan_state_map_value value;
							value.m_end_row = scan_end_iter->second.m_end_row;
							value.m_state = not_assigned;
							value.m_start_row_got = false;

							m_scan_state_map.insert(scan_end_iter, scan_state_map::value_type(end_row, value));
						}

						scan_state_map::iterator to_erase_iter = scan_end_iter;
						scan_end_row = to_erase_iter->second.m_end_row;
						++scan_end_iter;

						m_scan_state_map.erase(to_erase_iter);
					}

					// 确定扫描范围
					iter->second.m_end_row = scan_end_row;
					iter->second.m_state = scanning;

					std::string tablet_name = naming_strategy::tablet_name(
						meta.tablet_info_p->m_tablet_id, m_table_name,
						meta.tablet_info_p->m_start_row, meta.tablet_info_p->m_end_row
						);

					m_tablet_scan_info.m_tablet_name = tablet_name;
					m_tablet_scan_info.m_start_row = scan_start_row;
					m_tablet_scan_info.m_end_row = scan_end_row;

					// 在该范围建立一个scanner
					LOG(DEBUG, "create tablet_scanner on " << meta.tablet_info_p->m_table_name
						<< "[" << scan_start_row << "," << scan_end_row
						<< "]"
						);

					tablet_scanner_ptr tablet_scanner_p = tablet_scanner_ptr(
						new tablet_scanner(m_io_service_pool, ep, m_session_manager, m_tablet_scan_info)
						);

					tablet_scanner_p->async_get_next_row(
						this->get_lower_level_service().wrap(
						boost::bind(
						&this_type::on_get_next_row,
						this_ptr(),
						tablet_scanner_p,
						iter->second.m_start_row_got,
						meta.meta_got_time,
						_1, _2
						)
						)
						);

					++iter;
				}
			}

			if (iter != m_scan_state_map.end()
				&& iter->second.m_state == not_assigned)
			{
				// 在缓存中查找下一个子表
				m_location_cache.async_get_meta(
					this->get_lower_level_service().wrap(boost::bind(&this_type::on_get_meta, this_ptr(), _1, _2)),
					m_table_name, meta.tablet_info_p->m_end_row, false, 0
					);
			}
		}
		else
		{
			this->get_upper_level_service().dispatch(boost::bind(m_handler, row_result_ptr(), ec));
		}
	}

	void parallel_scanner::on_get_next_row(tablet_scanner_ptr tablet_scanner_p, bool start_row_got,
		uint64_t got_meta_time, row_result::ptr_type row_ptr, const error_code &ec)
	{
		if(!ec)
		{
			if (tablet_scanner_p->get_scan_start_row() != row_ptr->m_row_name
				|| !start_row_got)
			{
				this->get_upper_level_service().dispatch(boost::bind(m_handler, row_ptr, ec));
			}

			tablet_scanner_p->async_get_next_row(
				this->get_lower_level_service().wrap(
				boost::bind(
				&this_type::on_get_next_row,
				this_ptr(),
				tablet_scanner_p,
				start_row_got,
				got_meta_time,
				_1, _2
				)
				)
				);
		}
		else
		{
			if (ec == xbase::error::eos)
			{
				scan_state_map::iterator iter = m_scan_state_map.upper_bound(tablet_scanner_p->get_scan_start_row());
				--iter;

				iter->second.m_state = finished;

				LOG(DEBUG, "scan of " << m_table_name
					<< "[" << iter->first << "," << iter->second.m_end_row << "] finished, "
					<< tablet_scanner_p->get_returned_row_count() << " rows returned");

				// 检查是否所有的range都扫描完毕
				bool is_finished = true;
				iter = m_scan_state_map.begin();
				while (iter != m_scan_state_map.end())
				{
					if (iter->second.m_state != finished)
					{
						is_finished = false;
						break;
					}

					++iter;
				}

				if (is_finished)
				{
					// 扫描已经结束
					this->get_upper_level_service().dispatch(boost::bind(m_handler, row_result_ptr(), xbase::error::eos));
				}
			}
			else
			{
				if (ec == xbase::error::no_tablet_loaded)
				{
					this->get_upper_level_service().dispatch(boost::bind(m_handler, row_result_ptr(), ec));
				}
				else
				{
					unsigned returned_row_count = tablet_scanner_p->get_returned_row_count();

					LOG(DEBUG, "scan of " << m_table_name
						<< "[" << tablet_scanner_p->get_scan_start_row() << ","
						<< tablet_scanner_p->get_scan_end_row() << "] failed, "
						<< tablet_scanner_p->get_returned_row_count() << " rows returned. "
						<< "need to reload [" << tablet_scanner_p->get_cur_row() << ","
						<< tablet_scanner_p->get_scan_end_row() << "]");

					scan_state_map::iterator iter = m_scan_state_map.upper_bound(tablet_scanner_p->get_scan_start_row());
					--iter;

					if (returned_row_count > 0)
					{
						// 在扫描状态表中记录已经完成的部分
						scan_state_map_value value;
						value.m_end_row = iter->second.m_end_row;
						value.m_state = not_assigned;

						m_scan_state_map.insert(iter, scan_state_map::value_type(tablet_scanner_p->get_cur_row(), value));

						iter->second.m_state = finished;
						iter->second.m_end_row = tablet_scanner_p->get_cur_row();
					}
					else
					{
						iter->second.m_state = not_assigned;
					}

					m_location_cache.async_get_meta(
						this->get_lower_level_service().wrap(boost::bind(&this_type::on_get_meta, this_ptr(), _1, _2)),
						m_table_name, tablet_scanner_p->get_cur_row(), true, got_meta_time
						);
				}
			}
		}
	}
}
